﻿using Leo.Common.MQ;
using Leo.MSATracert;
using System.Runtime.CompilerServices;
using System.Text;

namespace Leo.Common.RabbitMQ
{

    public class RabbitConsumer : MQConsumerBase
    {
        private const ushort DefaultQos = 1;

        private readonly RabbitConnection connection;

        private readonly RabbitQueueInfo queue;

        private readonly int handlerCount;

        private readonly bool assignIndependentChannelToEachConsumer;

        public RabbitConsumer(RabbitConnection connection, RabbitQueueInfo queue, params IMessageHandler[] handlers) : base((handlers != null) ? Enumerable.ToList<IMessageHandler>(handlers) : null)
        {
            this.connection = connection;
            this.queue = queue;
            this.handlerCount = ((handlers != null) ? handlers.Length : 0);
            this.assignIndependentChannelToEachConsumer = connection.config.AssignIndependentChannelToEachConsumer;
        }

        protected override void ConnectToListen()
        {
            if (this.handlerCount <= 0)
            {
                Console.WriteLine("由于缺少处理程序，消费者未启动，订阅队列：" + this.queue.Name + "，handler数量：0");
                return;
            }
            object obj = this.assignIndependentChannelToEachConsumer ? this.connection.CreateNewChannel() : this.connection.GetChannel();
            ushort num = (this.queue.Qos > 0) ? this.queue.Qos : 1;
            object obj2 = obj;
            obj2.BasicQos(0U, num, false);
            EventingBasicConsumer eventingBasicConsumer = new EventingBasicConsumer(obj2);
            eventingBasicConsumer.Received += this.OnMessageReceived;
            IModelExensions.BasicConsume(obj2, this.queue.Name, false, eventingBasicConsumer);
            Console.WriteLine(string.Format("消费者已启动，订阅队列：{0}，Handler数量：{1}", this.queue.Name, this.handlerCount));
        }

        private void OnMessageReceived(object sender, BasicDeliverEventArgs e)
        {
            string @string = Encoding.UTF8.GetString(e.Body.ToArray());
            string type = e.BasicProperties.Type;
            EventingBasicConsumer eventingBasicConsumer = sender as EventingBasicConsumer;
            if (eventingBasicConsumer == null)
            {
                return;
            }
            List<IMessageHandler> matchHandlers = base.GetMatchHandlers(type);
            ReadOnlyMemory<byte> body = e.Body.ToArray();
            IBasicProperties properties = e.BasicProperties.CopyTo(eventingBasicConsumer.Model.CreateBasicProperties());
            RecvMsgInfo msg = new RecvMsgInfo(e.DeliveryTag, properties, body, eventingBasicConsumer.Model, e.Exchange, e.RoutingKey, @string);
            this.HandleAsync(matchHandlers, msg);
        }

        private Task HandleAsync(List<IMessageHandler> handlers, RecvMsgInfo msg)
        {
            RabbitConsumer.< HandleAsync > d__8 < HandleAsync > d__;

            < HandleAsync > d__.<> t__builder = AsyncTaskMethodBuilder.Create();

            < HandleAsync > d__.<> 4__this = this;

            < HandleAsync > d__.handlers = handlers;

            < HandleAsync > d__.msg = msg;

            < HandleAsync > d__.<> 1__state = -1;

            < HandleAsync > d__.<> t__builder.Start < RabbitConsumer.< HandleAsync > d__8 > (ref < HandleAsync > d__);
            return < HandleAsync > d__.<> t__builder.Task;
        }

        private void SetTraceId(IBasicProperties properties)
        {
            string traceId = Guid.NewGuid().ToString("N").ToLower();
            object obj;
            if (properties.Headers.TryGetValue("traceId", out obj))
            {
                byte[] array = obj as byte[];
                if (array != null)
                {
                    traceId = Encoding.UTF8.GetString(array);
                }
            }
            ServiceInvocationLinkTrace.GetPrincipal(traceId, null);
        }

        private void MessageDeadOrRetry(RecvMsgInfo msg)
        {
            MessageRetryStrategy retryStategy = this.queue.RetryStategy;
            int num = (retryStategy != null) ? retryStategy.RetryTimes : 0;
            MessageRetryStrategy retryStategy2 = this.queue.RetryStategy;
            TimeSpan timeSpan = (retryStategy2 != null) ? retryStategy2.RetryInterval : TimeSpan.FromMinutes(2.0);
            if (RabbitConsumer.GetDeathCount(msg.Properties.Headers) >= (long)num)
            {
                msg.Sender.BasicReject(msg.DeliveryTag, false);
                return;
            }
            msg.Sender.BasicAck(msg.DeliveryTag, false);
            msg.Properties.Expiration = ((long)timeSpan.TotalMilliseconds).ToString();
            PublicationAddress publicationAddress = RabbitConsumer.GetRetryAddress(msg.Properties.Headers) ?? RabbitConsumer.GetRetryAddress(this.queue.Properties);
            if (publicationAddress != null)
            {
                IModelExensions.BasicPublish(msg.Sender, publicationAddress, msg.Properties, msg.Body);
            }
        }

        //[return: Nullable(2)]
        private static PublicationAddress GetRetryAddress(IDictionary<string, object> headers)
        {
            if (headers == null || !headers.Any<KeyValuePair<string, object>>())
            {
                return null;
            }
            object obj;
            if (!headers.TryGetValue("retry-addr", out obj))
            {
                return null;
            }
            string text = obj as string;
            if (text != null && !string.IsNullOrEmpty(text))
            {
                return PublicationAddress.Parse(text);
            }
            byte[] array = obj as byte[];
            if (array == null)
            {
                return null;
            }
            return PublicationAddress.Parse(Encoding.UTF8.GetString(array));
        }

        private static long GetDeathCount(IDictionary<string, object> headers)
        {
            if (headers == null || !headers.Any<KeyValuePair<string, object>>())
            {
                return 0L;
            }
            object obj;
            if (!headers.TryGetValue("x-death", out obj))
            {
                return 0L;
            }
            List<object> list = obj as List<object>;
            if (list == null || !list.Any<object>())
            {
                return 0L;
            }
            Dictionary<string, object> dictionary = list.First<object>() as Dictionary<string, object>;
            if (dictionary == null)
            {
                return 0L;
            }
            object obj2;
            if (!dictionary.TryGetValue("count", ref obj2))
            {
                return 0L;
            }
            return (long)obj2;
        }


        private Task<bool> ExecuteSingleHandler(IMessageHandler handler, RecvMsgInfo msg)
        {
            RabbitConsumer.< ExecuteSingleHandler > d__13 < ExecuteSingleHandler > d__;

            < ExecuteSingleHandler > d__.<> t__builder = AsyncTaskMethodBuilder<bool>.Create();

            < ExecuteSingleHandler > d__.handler = handler;

            < ExecuteSingleHandler > d__.msg = msg;

            < ExecuteSingleHandler > d__.<> 1__state = -1;

            < ExecuteSingleHandler > d__.<> t__builder.Start < RabbitConsumer.< ExecuteSingleHandler > d__13 > (ref < ExecuteSingleHandler > d__);
            return < ExecuteSingleHandler > d__.<> t__builder.Task;
        }

        private Task<bool> ExecuteMultiHandler(IEnumerable<IMessageHandler> handlers, RecvMsgInfo msg)
        {
            RabbitConsumer.< ExecuteMultiHandler > d__14 < ExecuteMultiHandler > d__;

            < ExecuteMultiHandler > d__.<> t__builder = AsyncTaskMethodBuilder<bool>.Create();

            < ExecuteMultiHandler > d__.<> 4__this = this;

            < ExecuteMultiHandler > d__.handlers = handlers;

            < ExecuteMultiHandler > d__.msg = msg;

            < ExecuteMultiHandler > d__.<> 1__state = -1;

            < ExecuteMultiHandler > d__.<> t__builder.Start < RabbitConsumer.< ExecuteMultiHandler > d__14 > (ref < ExecuteMultiHandler > d__);
            return < ExecuteMultiHandler > d__.<> t__builder.Task;
        }


    }
}
